package org.qy.rocketmq.cs.rocket.consumer;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.qy.rocketmq.cs.rocket.listener.ConsumeMessageListener;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;

/**
 * All rights Reserved, Designed By www.cu-sc.com
 *
 * @Title:  OrderConsumer.java.java
 * @Package org.qy.rocketmq.cs.rocket.consumer
 * @Description: TODO
 * @Since: JDK 1.8
 * @Author: fangyukang
 * @Email: fangyk@cu-sc.com
 * @Version: v1.0.0
 * @Date: 2020/7/1 22:34
 * @Copyright: 2020 www.cu-sc.com All rights reserved. <br/>
 * 注意：本内容仅限于联通集团内部传阅，禁止外泄以及用于其他的商业目的<br/>
 */
@Slf4j
@Configuration
@ConfigurationProperties(prefix = "rocketmq.consumer")
public class OrderConsumer extends DefaultMQPushConsumer
{
    @Getter
    @Setter
    private Boolean enabled;

    @Getter
    @Setter
    private String subscribeTopic;

    @Getter
    @Setter
    private String subscribeTags;

    @Resource
    private ConsumeMessageListener messageListener;

    public OrderConsumer()
    {
        super();
    }

    @Override
    public String toString()
    {
        return "OrderConsumer{" +
                "enabled=" + enabled +
                ", subscribeTopic='" + subscribeTopic + '\'' +
                ", subscribeTags='" + subscribeTags + '\'' +
                ", messageListener=" + messageListener +
                "} " + super.toString();
    }

    @PostConstruct
    @Override
    public void start()
    {
        try {
            log.info(this.toString());
            if(Boolean.TRUE.equals(this.enabled)){
                Assert.notNull(messageListener, "rocketmq 消费 messageListener 不能为空，请实现IConsumeMessageListener接口类");
                this.registerMessageListener(messageListener);
                // 订阅消息, tags: 订单支付超时注销
                this.subscribe(subscribeTopic, subscribeTags);
                super.start();
                log.info("OrderConsumer|start|rocketmq consumer server 开启成功");
            }
        } catch (MQClientException e) {
            log.error("OrderConsumer|start|rocketmq consumer server 启动失败");
        }
    }

    @PreDestroy
    @Override
    public void shutdown()
    {
        if(Boolean.TRUE.equals(enabled))
        {
            super.shutdown();
            log.info("OrderConsumer|shutdown|rocketmq consumer server 关闭成功");
        }
    }
}
